package com.github.config;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;

import javax.annotation.PostConstruct;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.time.Duration;

@Configuration
public class RedisStreamConfig {
    @Value("${app.redis.stream-name}")
    private String streamKey;

    @Value("${app.redis.consumer-group}")
    private String consumerGroup;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;


    @PostConstruct
    public void initConsumerGroup() {
        try {
            // 检查并创建消费者组
            Boolean groupExists = stringRedisTemplate.opsForStream().groups(streamKey)
                    .stream()
                    .anyMatch(group -> consumerGroup.equals(group.groupName()));

            if (!groupExists) {
                stringRedisTemplate.opsForStream().createGroup(streamKey, ReadOffset.from("0-0"), consumerGroup);
                System.out.println("创建消费者组: " + consumerGroup);
            }
        } catch (Exception e) {
            System.out.println("消费者组可能已存在: " + e.getMessage());
        }
    }

    @Bean
    public StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> streamMessageListenerContainerOptions() {
        return StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                .builder()
                .pollTimeout(Duration.ofSeconds(1))
                .build();
    }


    @Bean
    public StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer(
            StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options,
            RedisConnectionFactory redisConnectionFactory,
            StreamListener<String, MapRecord<String, String, String>> streamListener) throws UnknownHostException {

        StreamMessageListenerContainer<String, MapRecord<String, String, String>> container =
                StreamMessageListenerContainer.create(redisConnectionFactory, options);

        String consumerName = InetAddress.getLocalHost().getHostName() + ":" + System.currentTimeMillis();

        // 订阅流
        Subscription subscription = container.receiveAutoAck(
                Consumer.from(consumerGroup, consumerName),
                StreamOffset.create(streamKey, ReadOffset.lastConsumed()),
                streamListener
        );

        container.start();
        System.out.println("Stream监听器已启动，消费者: " + consumerName);
        return container;
    }
}
